package com.thinking.kafkaTest;

import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.nio.charset.Charset;
import java.nio.charset.CharsetDecoder;
import java.util.List;

import org.apache.storm.spout.Scheme;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;

public class LogsScheme  implements Scheme{
	 public List<Object> deserialize(ByteBuffer ser) {
	        String msg = getStringByByteBuffer(ser);
	        return new Values(msg);
	    }

	    public Fields getOutputFields() {
	        return new Fields("pvanduvlogs");
	    }
	    /**
	     *
	     * ByteBuffer转换成字符串
	     *
	     * @param buffer
	     * @return
	     */
	    private String getStringByByteBuffer(ByteBuffer buffer) {
	        Charset charset = null;
	        CharsetDecoder decoder = null;
	        CharBuffer charBuffer = null;
	        try {
	            charset = Charset.forName("UTF-8");
	            decoder = charset.newDecoder();
	            charBuffer = decoder.decode(buffer.asReadOnlyBuffer());
	            return charBuffer.toString();
	        } catch (Exception e) {
	            return "";
	        }
	    }
}
